package com.bd08.flink.demo

import org.apache.flink.streaming.api.functions.source.SourceFunction

import java.sql.{ DriverManager}
case class Ke(pid:String,title:String,jie:String,price:String)

class MyJdbcSource extends SourceFunction[Ke]{
  override def run(sourceContext: SourceFunction.SourceContext[Ke]): Unit = {
    // JDBC
    Class.forName("com.mysql.cj.jdbc.Driver")
    val url ="jdbc:mysql://localhost:3306/kgc_db"
    val username = "root"
    val password = "password"
    val conn = DriverManager.getConnection(url,username,password)
    println(conn)
    val st = conn.createStatement()
    val rs = st.executeQuery("select * from kgc_base")
    while(rs.next()){
      val ke =Ke(rs.getString("pid"),
        rs.getString("title"),
        rs.getString("jie"),
        rs.getString("price"))
      println(ke)
      sourceContext.collect(ke)
      Thread.sleep(2000)
    }

  }

  override def cancel(): Unit = {


  }
}
